create function SplitFunction AS 'udtf.SplitFunction';
create function Split001Udf AS 'com.bigdata.udf.Split001Udf';
create function TjfromFormatUDF AS 'com.bigdata.udf.TjfromFormatUDF';

-------------------- source -------------------
------------- 58-topic -------------

CREATE TABLE hdp_lbg_supin_ods_zp_58_zzdetail_visitor (msg string)
with
(
'connector' = 'kafka',
'topic' = 'hdp_lbg_supin_ods_zp_58_zzdetail',
'properties.bootstrap.servers' = 'bh-kafka3-1.58dns.org:9092,bh-kafka3-2.58dns.org:9092,bh-kafka3-3.58dns.org:9092,bh-kafka3-4.58dns.org:9092,bh-kafka3-5.58dns.org:9092',
'properties.group.id' = 'hdp_lbg_supin_ods_zp_58_zzdetail_visitor',
'properties.client.id' = 'hdp_lbg_supin-hdp_lbg_supin_ods_zp_58_zzdetail-FRZfA',
'scan.startup.mode' = 'latest-offset',
'sink.parallelism' = '16',
'format' = 'raw'
);

------------- 赶集-topic -------------

CREATE TABLE hdp_lbg_zhaopin_ods_zp_ganji_zzdetail_visitor (msg string)
with
(
'connector' = 'kafka',
'topic' = 'hdp_lbg_supin_ods_zp_ganji_zzdetail',
'properties.bootstrap.servers' = 'bh-kafka3-1.58dns.org:9092,bh-kafka3-2.58dns.org:9092,bh-kafka3-3.58dns.org:9092,bh-kafka3-4.58dns.org:9092,bh-kafka3-5.58dns.org:9092',
'properties.group.id' = 'hdp_lbg_zhaopin-hdp_lbg_supin_ods_zp_ganji_zzdetail',
'properties.client.id' = 'hdp_lbg_zhaopin-hdp_lbg_supin_ods_zp_ganji_zzdetail-RsuDe',
'scan.startup.mode' = 'group-offsets',
'sink.parallelism' = '16',
'format' = 'raw'
);

------------- 兼职-topic -------------

CREATE TABLE hdp_lbg_zhaopin_ods_zp_jz_zzdetail_visitor (msg string)
with
(
'connector' = 'kafka',
'topic' = 'hdp_lbg_zhaopin_jianzhioffline_ods_58_zzdetail',
'properties.bootstrap.servers' = 'bh-kafka3-1.58dns.org:9092,bh-kafka3-2.58dns.org:9092,bh-kafka3-3.58dns.org:9092,bh-kafka3-4.58dns.org:9092,bh-kafka3-5.58dns.org:9092',
'properties.group.id' = 'hdp_lbg_supin-hdp_lbg_zhaopin_jianzhioffline_ods_58_zzdetail',
'properties.client.id' = 'hdp_lbg_supin-hdp_lbg_zhaopin_jianzhioffline_ods_58_zzdetail-P35vt',
'scan.startup.mode' = 'group-offsets',
'sink.parallelism' = '4',
'format' = 'raw'
);

------------- 同城宝-topic -------------
CREATE TABLE hdp_lbg_zhaopin_ods_zp_tongchengbao_zzdetail_visitor (msg string)
with
(
'connector' = 'kafka',
'topic' = 'hdp_lbg_ectech_lm_ods_click10',
'properties.bootstrap.servers' = '10.135.9.4:9092,10.135.9.5:9092,10.135.9.6:9092,10.135.9.7:9092,10.135.9.8:9092',
'properties.group.id' = 'hdp_lbg_supin-hdp_lbg_ectech_lm_ods_click10',
'properties.client.id' = 'hdp_lbg_supin-hdp_lbg_ectech_lm_ods_click10-zbFQB',
'scan.startup.mode' = 'group-offsets',
'sink.parallelism' = '4',
'format' = 'raw'
);

------------------- sink -------------------

CREATE TABLE hdp_lbg_supin_ods_zp_all_zzdetail_visitor
(
  buid bigint,
  cuid bigint,
  infoid bigint,
  sourcetype int,
  infoparams string,
  addtime bigint,
  tjfrom string,
  infodistype string
)
with
(
'connector' = 'kafka',
'topic' = 'hdp_lbg_supin_ods_zp_all_zzdetail_visitor',
'properties.bootstrap.servers' = 'bh-kafka3-1.58dns.org:9092,bh-kafka3-2.58dns.org:9092,bh-kafka3-3.58dns.org:9092,bh-kafka3-4.58dns.org:9092,bh-kafka3-5.58dns.org:9092',
'properties.client.id' = 'hdp_lbg_supin-hdp_lbg_supin_ods_zp_all_zzdetail_visitor-PQsgN',
'scan.startup.mode' = 'latest-offset',
'sink.parallelism' = '24',
'format' = 'json'
);

insert into hdp_lbg_supin_ods_zp_all_zzdetail_visitor
-------------- 58APP --------------

select
cast(arr[39] as bigint) as buid,
cast(arr[10] as bigint) as cuid,
cast(arr[38] as bigint) as infoid,
1 as sourcetype,
split_index(arr[35],'__',4) as infoparams, -- arr[47]
cast(arr[29] as bigint) as addtime,
arr[35] as tjfrom,
-- TjfromFormatUDF(arr[35]) as tjfrom2,
-- arr[46] as infodistype
case when split_index(arr[35],'__',3)='10119' then 'fangke'
  when split_index(arr[35],'__',3)='10077' then 'ppzq'
  when split_index(arr[35],'__',3)='10109' then 'superjob'
  when split_index(arr[35],'__',3)='hot' then 'huobao'
  when split_index(arr[35],'__',3)='-' then 'other'
  when split_index(arr[35],'__',3) in ('zhiding','10001','3') then 'zd'
  when split_index(arr[35],'__',3) in ('17','10063') then 'yx'
  when split_index(arr[35],'__',3) in ('17','10063') then 'yx'
  when split_index(arr[35],'__',3) in ('m','mqtt','10011') then 'tt'
  when split_index(arr[35],'__',3) in ('10062') then 'ptzb'
  when split_index(arr[35],'__',3) in ('0','p','b','bzp','r','tj','recj_add','PT','11') then 'pt'
  when split_index(arr[35],'__',3)='10061' then 'jpzb'
  when split_index(arr[35],'__',3) in ('zpjx','28','q','jx','zy','10042') then 'jz'
  when split_index(arr[35],'__',3) in ('j','10022') then 'jp'
  else split_index(arr[35],'__',3)
end as infodistype
from (
select Split001Udf(regex_str) as arr
from hdp_lbg_supin_ods_zp_58_zzdetail_visitor
,LATERAL TABLE(SplitFunction(msg,'\n'))
)t
where arr[30]='3' and cast(arr[10] as bigint)>0

union all
-------------- 58小程序 --------------

select
cast(arr[39] as bigint) as buid,
cast(arr[10] as bigint) as cuid,
cast(arr[38] as bigint) as infoid,
6 as sourcetype,
split_index(arr[35],'__',4) as infoparams, -- arr[47]
cast(arr[29] as bigint) as addtime,
arr[35] as tjfrom,
-- TjfromFormatUDF(arr[35]) as tjfrom2,
-- arr[46] as infodistype
case when split_index(arr[35],'__',3)='10119' then 'fangke'
  when split_index(arr[35],'__',3)='10077' then 'ppzq'
  when split_index(arr[35],'__',3)='10109' then 'superjob'
  when split_index(arr[35],'__',3)='hot' then 'huobao'
  when split_index(arr[35],'__',3)='-' then 'other'
  when split_index(arr[35],'__',3) in ('zhiding','10001','3') then 'zd'
  when split_index(arr[35],'__',3) in ('17','10063') then 'yx'
  when split_index(arr[35],'__',3) in ('17','10063') then 'yx'
  when split_index(arr[35],'__',3) in ('m','mqtt','10011') then 'tt'
  when split_index(arr[35],'__',3) in ('10062') then 'ptzb'
  when split_index(arr[35],'__',3) in ('0','p','b','bzp','r','tj','recj_add','PT','11') then 'pt'
  when split_index(arr[35],'__',3)='10061' then 'jpzb'
  when split_index(arr[35],'__',3) in ('zpjx','28','q','jx','zy','10042') then 'jz'
  when split_index(arr[35],'__',3) in ('j','10022') then 'jp'
  else split_index(arr[35],'__',3)
end as infodistype
from (
select Split001Udf(regex_str) as arr
from hdp_lbg_supin_ods_zp_58_zzdetail_visitor
,LATERAL TABLE(SplitFunction(msg,'\n'))
)t
where arr[30]='7' and cast(arr[10] as bigint)>0
-------------- 58PC --------------

union all
select
cast(arr[39] as bigint) as buid,
cast(arr[10] as bigint) as cuid,
cast(arr[38] as bigint) as infoid,
11 as sourcetype,
regexp_replace(split_index(arr[35],'__',4),'%2C',',') as infoparams, -- arr[47]
cast(arr[29] as bigint) as addtime,
arr[35] as tjfrom,
-- TjfromFormatUDF(arr[35]) as tjfrom2,
-- arr[46] as infodistype
case when split_index(arr[35],'__',3)='10119' then 'fangke'
  when split_index(arr[35],'__',3)='10077' then 'ppzq'
  when split_index(arr[35],'__',3)='10109' then 'superjob'
  when split_index(arr[35],'__',3)='hot' then 'huobao'
  when split_index(arr[35],'__',3)='-' then 'other'
  when split_index(arr[35],'__',3) in ('zhiding','10001','3') then 'zd'
  when split_index(arr[35],'__',3) in ('17','10063') then 'yx'
  when split_index(arr[35],'__',3) in ('17','10063') then 'yx'
  when split_index(arr[35],'__',3) in ('m','mqtt','10011') then 'tt'
  when split_index(arr[35],'__',3) in ('10062') then 'ptzb'
  when split_index(arr[35],'__',3) in ('0','p','b','bzp','r','tj','recj_add','PT','11') then 'pt'
  when split_index(arr[35],'__',3)='10061' then 'jpzb'
  when split_index(arr[35],'__',3) in ('zpjx','28','q','jx','zy','10042') then 'jz'
  when split_index(arr[35],'__',3) in ('j','10022') then 'jp'
  else split_index(arr[35],'__',3)
end as infodistype
from (
select Split001Udf(regex_str) as arr
from hdp_lbg_supin_ods_zp_58_zzdetail_visitor
,LATERAL TABLE(SplitFunction(msg,'\n'))
)t
where arr[30]='1' and cast(arr[10] as bigint)>0
union all
-------------- 赶集APP --------------

select
cast(arr[39] as bigint) as buid,
cast(arr[10] as bigint) as cuid,
cast(arr[38] as bigint) as infoid,
2 as sourcetype,
split_index(arr[35],'__',4) as infoparams, -- arr[47]
cast(arr[29] as bigint) as addtime,
arr[35] as tjfrom,
-- TjfromFormatUDF(arr[35]) as tjfrom2,
-- arr[46] as infodistype
case when split_index(arr[35],'__',3)='10119' then 'fangke'
  when split_index(arr[35],'__',3)='10077' then 'ppzq'
  when split_index(arr[35],'__',3)='10109' then 'superjob'
  when split_index(arr[35],'__',3)='hot' then 'huobao'
  when split_index(arr[35],'__',3)='-' then 'other'
  when split_index(arr[35],'__',3) in ('zhiding','10001','3') then 'zd'
  when split_index(arr[35],'__',3) in ('17','10063') then 'yx'
  when split_index(arr[35],'__',3) in ('17','10063') then 'yx'
  when split_index(arr[35],'__',3) in ('m','mqtt','10011') then 'tt'
  when split_index(arr[35],'__',3) in ('10062') then 'ptzb'
  when split_index(arr[35],'__',3) in ('0','p','b','bzp','r','tj','recj_add','PT','11') then 'pt'
  when split_index(arr[35],'__',3)='10061' then 'jpzb'
  when split_index(arr[35],'__',3) in ('zpjx','28','q','jx','zy','10042') then 'jz'
  when split_index(arr[35],'__',3) in ('j','10022') then 'jp'
  else split_index(arr[35],'__',3)
end as infodistype
from (
select Split001Udf(regex_str) as arr
from hdp_lbg_zhaopin_ods_zp_ganji_zzdetail_visitor
,LATERAL TABLE(SplitFunction(msg,'\n'))
)t
where arr[30]='3' and cast(arr[10] as bigint)>0 and split_index(arr[35],'__',0) not like 'mini_ganji_%'
union all
-------------- 赶集小程序 --------------

select
cast(arr[39] as bigint) as buid,
cast(arr[10] as bigint) as cuid,
cast(arr[38] as bigint) as infoid,
7 as sourcetype,
split_index(arr[35],'__',4) as infoparams, -- arr[47]
cast(arr[29] as bigint) as addtime,
arr[35] as tjfrom,
-- TjfromFormatUDF(arr[35]) as tjfrom2,
-- arr[46] as infodistype
case when split_index(arr[35],'__',3)='10119' then 'fangke'
  when split_index(arr[35],'__',3)='10077' then 'ppzq'
  when split_index(arr[35],'__',3)='10109' then 'superjob'
  when split_index(arr[35],'__',3)='hot' then 'huobao'
  when split_index(arr[35],'__',3)='-' then 'other'
  when split_index(arr[35],'__',3) in ('zhiding','10001','3') then 'zd'
  when split_index(arr[35],'__',3) in ('17','10063') then 'yx'
  when split_index(arr[35],'__',3) in ('17','10063') then 'yx'
  when split_index(arr[35],'__',3) in ('m','mqtt','10011') then 'tt'
  when split_index(arr[35],'__',3) in ('10062') then 'ptzb'
  when split_index(arr[35],'__',3) in ('0','p','b','bzp','r','tj','recj_add','PT','11') then 'pt'
  when split_index(arr[35],'__',3)='10061' then 'jpzb'
  when split_index(arr[35],'__',3) in ('zpjx','28','q','jx','zy','10042') then 'jz'
  when split_index(arr[35],'__',3) in ('j','10022') then 'jp'
  else split_index(arr[35],'__',3)
end as infodistype
from (
select Split001Udf(regex_str) as arr
from hdp_lbg_zhaopin_ods_zp_ganji_zzdetail_visitor
,LATERAL TABLE(SplitFunction(msg,'\n'))
)t
where arr[30]='3' and cast(arr[10] as bigint)>0 and split_index(arr[35],'__',0) like 'mini_ganji_%'

union all
-------------- 兼职APP --------------

select
cast(arr[39] as bigint) as buid,
cast(arr[10] as bigint) as cuid,
cast(arr[38] as bigint) as infoid,
4 as sourcetype,
split_index(arr[35],'__',4) as infoparams, -- arr[47]
cast(arr[29] as bigint) as addtime,
arr[35] as tjfrom,
-- TjfromFormatUDF(arr[35]) as tjfrom2,
-- arr[46] as infodistype
case when split_index(arr[35],'__',3)='10119' then 'fangke'
  when split_index(arr[35],'__',3)='10077' then 'ppzq'
  when split_index(arr[35],'__',3)='10109' then 'superjob'
  when split_index(arr[35],'__',3)='hot' then 'huobao'
  when split_index(arr[35],'__',3)='-' then 'other'
  when split_index(arr[35],'__',3) in ('zhiding','10001','3') then 'zd'
  when split_index(arr[35],'__',3) in ('17','10063') then 'yx'
  when split_index(arr[35],'__',3) in ('17','10063') then 'yx'
  when split_index(arr[35],'__',3) in ('m','mqtt','10011') then 'tt'
  when split_index(arr[35],'__',3) in ('10062') then 'ptzb'
  when split_index(arr[35],'__',3) in ('0','p','b','bzp','r','tj','recj_add','PT','11') then 'pt'
  when split_index(arr[35],'__',3)='10061' then 'jpzb'
  when split_index(arr[35],'__',3) in ('zpjx','28','q','jx','zy','10042') then 'jz'
  when split_index(arr[35],'__',3) in ('j','10022') then 'jp'
  else split_index(arr[35],'__',3)
end as infodistype
from (
select Split001Udf(regex_str) as arr
from hdp_lbg_zhaopin_ods_zp_jz_zzdetail_visitor
,LATERAL TABLE(SplitFunction(msg,'\n'))
)t
where arr[30]='3' and cast(arr[10] as bigint)>0
union all
-------------- 同城宝小程序 8 --------------

select
cast(arr[27] as bigint) as buid,
cast(regexp_extract(arr[9],'cuid":"(.*?)"', 1) as bigint) as cuid,
cast(arr[29] as bigint) as infoid,
8 as sourcetype,
split_index(regexp_extract(arr[9],'tjfrom":"(.*?)"', 1),'__',4) as infoparams, -- arr[47]
cast(arr[13] as bigint) as addtime,
regexp_extract(arr[9],'tjfrom":"(.*?)"', 1) as tjfrom,
-- TjfromFormatUDF(arr[35]) as tjfrom2,
-- arr[46] as infodistype
case when split_index(regexp_extract(arr[9],'tjfrom":"(.*?)"', 1),'__',3)='10119' then 'fangke'
  when split_index(regexp_extract(arr[9],'tjfrom":"(.*?)"', 1),'__',3)='10077' then 'ppzq'
  when split_index(regexp_extract(arr[9],'tjfrom":"(.*?)"', 1),'__',3)='10109' then 'superjob'
  when split_index(regexp_extract(arr[9],'tjfrom":"(.*?)"', 1),'__',3)='hot' then 'huobao'
  when split_index(regexp_extract(arr[9],'tjfrom":"(.*?)"', 1),'__',3)='-' then 'other'
  when split_index(regexp_extract(arr[9],'tjfrom":"(.*?)"', 1),'__',3) in ('zhiding','10001','3') then 'zd'
  when split_index(regexp_extract(arr[9],'tjfrom":"(.*?)"', 1),'__',3) in ('17','10063') then 'yx'
  when split_index(regexp_extract(arr[9],'tjfrom":"(.*?)"', 1),'__',3) in ('17','10063') then 'yx'
  when split_index(regexp_extract(arr[9],'tjfrom":"(.*?)"', 1),'__',3) in ('m','mqtt','10011') then 'tt'
  when split_index(regexp_extract(arr[9],'tjfrom":"(.*?)"', 1),'__',3) in ('10062') then 'ptzb'
  when split_index(regexp_extract(arr[9],'tjfrom":"(.*?)"', 1),'__',3) in ('0','p','b','bzp','r','tj','recj_add','PT','11') then 'pt'
  when split_index(regexp_extract(arr[9],'tjfrom":"(.*?)"', 1),'__',3)='10061' then 'jpzb'
  when split_index(regexp_extract(arr[9],'tjfrom":"(.*?)"', 1),'__',3) in ('zpjx','28','q','jx','zy','10042') then 'jz'
  when split_index(regexp_extract(arr[9],'tjfrom":"(.*?)"', 1),'__',3) in ('j','10022') then 'jp'
  else split_index(regexp_extract(arr[9],'tjfrom":"(.*?)"', 1),'__',3)
end as infodistype
from (
select Split001Udf(regex_str) as arr
from hdp_lbg_zhaopin_ods_zp_tongchengbao_zzdetail_visitor
,LATERAL TABLE(SplitFunction(msg,'\n'))
)t
where arr[9] like '%tjfrom%' and arr[9] like '%cuid%'
and regexp_extract(arr[9],'businessLine":"(.*?)"', 1)='job'
and cast(regexp_extract(arr[9],'cuid":"(.*?)"', 1) as bigint)>0
;


